#include "dbenv.hxx"
#include "datastore.hxx"
#include <iostream>
#include <sstream>
#include <assert.h>
#include "util/common.hxx"
#include "dbt.hxx"

DbEnv* env = NULL; 
DataStore* store = NULL;

std::string mymapfunc(Dbt& key) {

    return "mypartition";
}    

int sequentialWrite(DbEnv* env, DataStore* store, int start, int end) {

    DbStatus ret;

    std::string value(128, 'y'); Dbt valuedbt(value);
    for(int i = start; i <= end; ++i) {
   
        std::ostringstream oss; oss<<i; 
        std::string key = oss.str(); Dbt keydbt(key);
        ret = store->put(keydbt, valuedbt);
        if(!ret.isok())  {std::cout<<"put failed"<<ret.msg(); return -1;}
    }


    return 0;
}

int sequentialRead(DbEnv* env, DataStore* store, int start, int end) {

    for(int i = start; i <= end; ++i) {
   
        std::ostringstream oss; oss<<i;
        std::string key = oss.str();
        Dbt keydbt(key), valuedbt;
        DbStatus ret = store->get(keydbt, valuedbt);
       
        if(ret.getCode() == DbStatus::DB_KEYNOTFOUND) {
            std::cout<<i<<" "<<ret.msg()<<"\n"; 
        }
        else {

            std::string value0(128, 'y');
            if(value0 != valuedbt.toString()) {
                std::cout<<i<<" not matched: "<<std::endl;
            }
        }
    }
}

int N = 10000;

void * thread_write(void *arg) {

    int start = reinterpret_cast<long>(arg);
    int end = start + N;

    sequentialWrite(env, store, start, end);

    return NULL;
}

void startWriteThreads(int n) {

    pthread_t id[128];
    for(int i = 0; i < n; ++i) {
        id[i] = utility::startThread(thread_write, reinterpret_cast<void*>(i*N));
    }

    for(int i = 0; i < n; ++i) {
        utility::waitThread(id[i]);
    }
}


void * thread_read(void *arg) {
    int start = reinterpret_cast<long>(arg);
    int end = start + N;

    sequentialRead(env, store, start, end);

    return NULL;
}

void startReadThreads(int n) {
    pthread_t id[128];
    for(int i = 0; i < n; ++i) {
        id[i] = utility::startThread(thread_read, reinterpret_cast<void*>(i*N));
    }

    for(int i = 0; i < n; ++i) {
        utility::waitThread(id[i]);
    }
}


void startReadWriteThreads(int n) {
    
    pthread_t idr[128], idw[128];
    for(int i = 0; i < n; ++i) {
    
        idw[i] = utility::startThread(thread_write, reinterpret_cast<void*>(i*N));
        idr[i] = utility::startThread(thread_read, reinterpret_cast<void*>(i*N));
    }

    for(int i = 0; i < n; ++i) {
        
        utility::waitThread(idr[i]);
        utility::waitThread(idw[i]);
    }
}




int main(int argc, char** argv) {

    std::string logmsgFile = argv[1];
    logmsgFile = logmsgFile + "/" + "mydb.msg";

    env = new DbEnv(argv[1], argv[1], logmsgFile, 10*1024*1024);
    DbStatus ret = env->open();
    if(!ret.isok()) {std::cout<<"env cannot be opened"<<ret.msg(); return -1;}

    store = new DataStore(*env);
    const std::string dsName("mystore");
    uint32_t cacheSize = 1000000; 
    uint32_t bufferSize = 10000; 
    uint32_t indexArrayCapacity = 64*1024; 
    uint32_t segCapacity = 32*1024*1024; 
    MAPPINGFUNC* mapFunc = mymapfunc;
    ret = store->open(dsName, cacheSize, indexArrayCapacity, segCapacity,
            mapFunc, 0);
    if(!ret.isok())  {std::cout<<"open"<<ret.msg(); return -1;}

    ret = store->addPartition("mypartition");
    if(!ret.isok())  {std::cout<<"add partition"<<ret.msg(); return -1;}

    ret = env->start(); 
    if(!ret.isok())  {std::cout<<"env start"<<ret.msg(); return -1;}

    if(argv[2][0] == '0') {
        std::cout<<"concurrent write mode\n";
        startWriteThreads(3);


        std::cout<<"concurrent reading mode\n";
        startReadThreads(3);
    }
    else if(argv[2][0] == '1') {
        std::cout<<"readwrite mode\n";
        startReadWriteThreads(3);
    }

    env->flushlog();

    store->flush();

    delete store;
    delete env;
}

